package com.rocketmq.one.message;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class TagMessage {

    @Test
    public void tagProduce() throws Exception {
        // 获取生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("tag-produce");
        // 设置服务对象
        producer.setNamesrvAddr("8.134.163.217:9876");
        // 开始
        producer.start();
        // 处理
        Message msg = new Message("TopicTest", "tag-A","我是一个标签tag-A消息".getBytes());
        Message msg1 = new Message("TopicTest", "tag-B","我是一个标签tag-B消息".getBytes());
        // 发送
        producer.send(Arrays.asList(msg,msg1));
        // 关闭
        producer.shutdown();

    }
    @Test
    public void tagAConsume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consume-a");
        consumer.setNamesrvAddr("8.134.163.217:9876");
        consumer.subscribe("TopicTest", "tag-A");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            System.out.println("消息内容："+new String(list.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
    @Test
    public void tagAllConsume() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consume-all");
        consumer.setNamesrvAddr("8.134.163.217:9876");
        consumer.subscribe("TopicTest", "tag-A || tag-B");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            System.out.println("当前线程 --- "+Thread.currentThread().getId());
            System.out.println("消息内容："+new String(list.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
